/**
 * 2017年6月29日
 */
package org.textplantform.common.core.mq;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.textplantform.common.Constant;
import org.textplantform.common.api.mq.PlantformTextProducer;
import org.textplantform.common.api.mq.PlantformTextProducerFactory;

/**
 * @author Alex
 *
 */
public class KafkaProducerFactory implements PlantformTextProducerFactory<ProducerRecord<String, String>,Future<RecordMetadata>,Callback,Producer<String, String>> {
	private KafkaProducerFactory(){}
	private static Properties props = new Properties();
	static {
		try {
			props.load(KafkaProducerFactory.class.getClassLoader().getResourceAsStream("meta/mq-producer.properties"));
		} catch (IOException e) {
			throw new RuntimeException(e.getMessage(),e);
		}
	}
	private static KafkaProducerFactory inst = null;
	public static KafkaProducerFactory getInstance(){
		if(inst == null){
			synchronized (KafkaProducerFactory.class) {
				if(inst == null)
					inst = new KafkaProducerFactory();
			}
		}
		return inst;
	}
	/**
	 * 官方文档建议是多线程共享一个Producer对象.
	 */
	private static Producer<String, String> underLayingProducer = new KafkaProducer<String, String>(props);
	@Override
	public PlantformTextProducer<ProducerRecord<String, String>, Future<RecordMetadata>, Callback,Producer<String, String>> getProducer() {
		return new PlantformKafkaProducer(underLayingProducer);
	}
	public PlantformKafkaProducer getKafkaProducer() {
		return new PlantformKafkaProducer(underLayingProducer);
	}
	
	public static void main(String[] args) {
		KafkaProducerFactory kpf = KafkaProducerFactory.getInstance();
		PlantformKafkaProducer p = kpf.getKafkaProducer();
		for(int i=0;i<10;i++){
			p.delegate.send(new ProducerRecord<String, String>(Constant.Values.MQ_TOPIC, String.valueOf(i), String.valueOf(i)));
		}
		p.delegate.close();
		
		
	}
}
